package com.yss.netty.future;

import com.yss.netty.protocol.Message;

import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * Response future.
 */
public class ResponseFuture {

    /**
     * Good response
     */
    private boolean isOk;

    /**
     * The cause of error
     */
    private String errCause;

    /**
     * Receive message
     */
    private Message message;

    /**
     * Blocking current thread
     */
    private CountDownLatch latch;

    /**
     * Read response timeout event
     * The unit is milliseconds
     */
    private long waitTimeOut;

    /**
     * Callback for async send message
     */
    private ResponseCallback callback;

    /**
     * Timestamp of sending data
     */
    private final long beginTimeStamp;

    /**
     * Request opaque mapping
     */
    private static final ConcurrentHashMap<Long, ResponseFuture> responseFutures;

    static {
        responseFutures = new ConcurrentHashMap<>();
    }

    public ResponseFuture(final long waitTimeOut) {
        this.waitTimeOut = waitTimeOut;
        this.latch = new CountDownLatch(1);
        beginTimeStamp = System.currentTimeMillis();
    }

    /**
     * Add a future
     *
     * @param opaque
     */
    public void addFuture(long opaque) {
        responseFutures.put(opaque, this);
    }

    /**
     * Remove a future
     *
     * @param opaque
     */
    public void removeFuture(long opaque) {
        responseFutures.remove(opaque);
    }

    public static ResponseFuture getResponseFuture(long opaque) {
        return responseFutures.get(opaque);
    }

    /**
     * Put response
     *
     * @param message Response from other server
     */
    public void putResponse(final Message message) {
        this.message = message;
        latch.countDown();
        if (message != null) {
            this.removeFuture(message.getOpaque());
        }
    }

    /**
     * Get response
     * <p>
     * Note:
     * if response is null, the current thread will be blocking
     *
     * @return
     */
    public Message waitResponse() {
        try {
            // avoid waiting
            latch.await(this.waitTimeOut, TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        return this.message;
    }

    public boolean isOk() {
        return isOk;
    }

    public void setOk(boolean ok) {
        isOk = ok;
    }

    public String getErrCause() {
        return errCause;
    }

    public void setErrCause(String errCause) {
        this.errCause = errCause;
    }

    public Message getMessage() {
        return message;
    }

    public void setMessage(Message message) {
        this.message = message;
    }

    public ResponseCallback getCallback() {
        return callback;
    }

    public void setCallback(ResponseCallback callback) {
        this.callback = callback;
    }

    public void executeCallback(final Message message) {
        callback.accept(message);
        removeFuture(message.getOpaque());
    }

    public long getBeginTimeStamp() {
        return beginTimeStamp;
    }

    public long getWaitTimeOut() {
        return waitTimeOut;
    }

    public static ConcurrentHashMap<Long, ResponseFuture> getResponseFutures() {
        return responseFutures;
    }

    /**
     * Note:
     * Consider the following situation:
     * If the data is sent successfully, but other services are down.
     * It may cause {@link ResponseFuture#responseFutures} OOM
     */
    public static void scanTimeOutFutures() {
        Iterator<Map.Entry<Long, ResponseFuture>> iterator = responseFutures.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry<Long, ResponseFuture> next = iterator.next();
            ResponseFuture responseFuture = next.getValue();
            if ((responseFuture.getBeginTimeStamp() + responseFuture.getWaitTimeOut() + 1000L) < System.currentTimeMillis()) {
                // time out should be clear.
                iterator.remove();
            }
        }
    }

    @Override
    public String toString() {
        return "ResponseFuture{" +
                "isOk=" + isOk +
                ", errCause='" + errCause + '\'' +
                ", message=" + message +
                ", latch=" + latch +
                ", waitTimeOut=" + waitTimeOut +
                '}';
    }
}
